package com.zw.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * @author morningstar
 * @date 2021/11/30
 * @desc
 */
public class MyConsumer {

    public final static String TOPIC_NAME = "my-replicated-topic";
    public final static String CONSUMER_GROUP_NAME = "testGroup";

    public static void main(String[] args) {

        // 1.设置参数
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "tajlmt0law.cogiot.net:39092,tajlmt0law.cogiot.net:39093,tajlmt0law.cogiot.net:39094");

        // 消费分组名
        props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);


        /**
         * 自动提交
         */
        // 是否⾃动提交offset，默认就是true
//        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        // ⾃动提交offset的间隔时间
//        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

        /**
         * 手动提交
         */
        // 是否⾃动提交offset，默认就是true
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        /**
         * 当消费者是一个新的消费主题下的新的消费组，或者指定offset消费方式时offset不存在
         * latest(默认): 只消费自己启动之后发送到主题的消息
         * earliest: 第一次从头开始消费。以后按照offset记录继续消费，区别于 consumer.seekToBeginning() 每次都是从头开始消费
         */
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // consumer给broker发送⼼跳的间隔时间 1s
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
        // kafka如果超过10秒没有收到消费者的⼼跳，则会把消费者踢出消费组，通过rebalance机制把分区分配给其他消费者。
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);

        // ⼀次poll最⼤拉取消息的条数，可以根据消费速度的快慢来设置
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        //如果两次poll的时间如果超出了30s的时间间隔，kafka会认为其消费能⼒过弱，将其踢出消费组。将分区分配给其他消费者。-rebalance
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);


        // 把发送的key从字符串序列化为字节数组
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 把发送消息value从字符串序列化为字节数组
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

        // 消费者订阅主题
//        consumer.subscribe(Arrays.asList(TOPIC_NAME));

        // 消费者指定分区
//        consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME,0)));

        // 消息回溯消费
//        consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME,0)));
//        consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME,0)));

        // 指定offset消费
//        consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME,0)));
//        consumer.seek(new TopicPartition(TOPIC_NAME,0),10);

        // 从指定时间开始消费
        List<PartitionInfo> topicPartitions = consumer.partitionsFor(TOPIC_NAME);
        // 从1⼩时前开始消费
        long fetchDataTime = new Date().getTime() - 1000 * 60 * 60;

        Map<TopicPartition, Long> map = new HashMap<>();

        for (PartitionInfo par : topicPartitions) {
            map.put(new TopicPartition(TOPIC_NAME, par.partition()), fetchDataTime);
        }

        Map<TopicPartition, OffsetAndTimestamp> parMap = consumer.offsetsForTimes(map);
        for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : parMap.entrySet()) {
            TopicPartition key = entry.getKey();
            OffsetAndTimestamp value = entry.getValue();
            if (key == null || value == null) continue;
            Long offset = value.offset();
            System.out.println("partition-" + key.partition() + "|offset-" + offset);
            System.out.println();
            // 根据消费⾥的timestamp确定offset
            if (value != null) {
                consumer.assign(Arrays.asList(key));
                consumer.seek(key, offset);
            }
        }

        while (true) {
            /**
             * poll() API 是拉取消息的长轮询
             */
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord record : records) {
                //4.打印消息
                System.out.printf("收到消息：partition = %d,offset = %d, key = %s, value = %s%n", record.partition(),
                        record.offset(), record.key(), record.value());
            }

            //所有的消息已消费完
            if (records.count() > 0) {
                // 有消息
                /**
                 * ⼿动同步提交offset，当前线程会阻塞直到offset提交成功，⼀般使⽤同步提交，因为提交之后⼀般也没有什么逻辑代码了
                 */
//                consumer.commitSync();  // =======阻塞=== 提交成功


                /**
                 * ⼿动异步提交offset，当前线程提交offset不会阻塞，可以继续处理后⾯的程序逻辑
                 */
                /*consumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                        if (exception != null) {
                            System.err.println("Commit failed for " + offsets);
                            System.err.println("Commit failed exception: " + exception.getStackTrace());
                        }
                    }
                });*/

            }

        }

    }
}
